package com.huatai.datacenter.service;

import com.alibaba.fastjson.JSONObject;
import com.huatai.datacenter.config.KafkaConfig;
import com.huatai.datacenter.entity.ClusterEntity;
import com.huatai.datacenter.entity.TopicTaskEntity;
import com.huatai.datacenter.entity.datamodel.DataModelFieldEntity;
import com.huatai.datacenter.job.KafkaConsumerListener;
import com.huatai.datacenter.job.KafkaListenerJob;
import com.huatai.datacenter.mapper.ClusterMapper;
import com.huatai.datacenter.mapper.TopicTaskMapper;
import com.huatai.dataservice.dubbo.DataDisposeDubboAPI;
import org.apache.catalina.connector.ClientAbortException;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;

/**
 * @author Lion
 * @date 2023/6/2  13:43
 */
@Service
public class TopicTaskService  extends BaseService<TopicTaskMapper, TopicTaskEntity> {

	@Autowired
	private KafkaConsumerService kafkaConsumerService;

	@Autowired
	private ClusterService clusterService;

	@Autowired
	private KafkaConsumerListener kafkaConsumerListener;

	@Autowired
	private DataModelApplyService dataModelApplyService;

	@Autowired
	private DataModelFieldService dataModelFieldService;

	@DubboReference
	private DataDisposeDubboAPI dataDisposeDubbo;

	@Autowired
	private RabbitMQService rabbitMQService;

	/**
	 * @author Lion
	 * @description 保存主题任务并启动线程监听
	 * @param entity
	 * @return boolean
	 */
	public void savaTopicTask(TopicTaskEntity entity){

		saveTopicTaskData(entity);
	}

	/**
	 * @author Lion
	 * @description 更新主题任务
	 * @param  entity
	 * @return void
	 */
	@Transactional
	public void updateTopicTask(TopicTaskEntity entity){

		List list = Arrays.asList(entity.getId());
		deleteTopicTaskById(list);
		if ("kafka".equals(entity.getTypeMq())){
			saveTopicTaskData(entity);
		}else if ("rabbitmq".equals(entity.getTypeMq())){
			rabbitMQService.createRabbitMqTask(entity);
		}
	}

	/**
	 * @author Lion
	 * @description 获取线程
	 * @param  threadId
	 * @return java.lang.Thread
	 */
	public Thread findThread(long threadId) {
		ThreadGroup group = Thread.currentThread().getThreadGroup();
		while(group != null) {
			Thread[] threads = new Thread[(int)(group.activeCount() * 1.2)];
			int count = group.enumerate(threads, true);
			for(int i = 0; i < count; i++) {
				if(threadId == threads[i].getId()) {
					return threads[i];
				}
			}
			group = group.getParent();
		}
		return null;
	}



	/**
	 * @author Lion
	 * @description 根据主题任务id关闭线程并删除任务数据
	 * @param  @param id
	 * @return void
	 */
	public void deleteTopicTaskById(List ids){
		Map<String, Object> map = RabbitMQService.map;
		if (ids != null){
			List<TopicTaskEntity> list = this.listByIds(ids);
			for (TopicTaskEntity topicTaskEntity : list) {
				if ("rabbitmq".equals(topicTaskEntity.getTypeMq())){
					Long id = (Long) map.get(topicTaskEntity.getId().toString());
					//再删除任务数据
					this.removeById(topicTaskEntity.getId());
					Thread s = this.findThread(id);
					try {
						s.interrupt();
					}
					catch (Exception e){

					}
				}else {
					//for (int i=0;i<ids.size();i++){
						/*long id = Long.valueOf(ids.get(i).toString());
						//先关闭线程
						long threadId = this.getById(id).getThreadId();*/
						Long threadId = topicTaskEntity.getThreadId();
						Thread s = this.findThread(threadId);
						//再删除任务数据
						this.removeById(topicTaskEntity.getId());
						try {
							s.interrupt();
						}
						catch (Exception e){

						}
					//}
				}
			}
		}
	}



	/**
	 * @author Lion
	 * @description 创建监听线程并保存主题任务数据
	 * @param  entity
	 * @return void
	 */
	@Transactional
	public void saveTopicTaskData(TopicTaskEntity entity){

		KafkaConfig kafkaConfig = new KafkaConfig();
		kafkaConfig.compareList(this.list());

		List<DataModelFieldEntity> dataModelFieldEntities = dataModelFieldService.getDataModelFieldList(entity.getModelId());
		dataDisposeDubbo.existTable(entity.getDataTable(),jsonStructureData(dataModelFieldEntities));

		String broker = clusterService.selectById(entity.getClusterId()).getBroker();
		Properties consumerProps =kafkaConsumerService.generatorConsumerProps(broker,entity.getGroupId(),true);
		consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,100);

		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(consumerProps);
		// 订阅消息
		kafkaConsumer.subscribe(Collections.singletonList(entity.getTopicName()));
		//启动消息监听线程
		KafkaListenerJob kafkaListenerJob = new KafkaListenerJob(kafkaConsumerListener,kafkaConsumer,entity.getModelId(),entity.getModelName(),entity.getDataTable(),entity.getStructure(),entity.getQualityRule());
		Thread t = new Thread(kafkaListenerJob);
		t.start();
		entity.setThreadId(t.getId());

		//判断模型id和质量是否存在，如果存在，则写入一条记录到记录表中
		super.save(entity);
		saveUseModelData(entity);

	}

	/**
	 * @author Lion
	 * @description 判断是否选择模型，如果有则写入
	 * @param entity
	 * @return void
	 */
	public void saveUseModelData(TopicTaskEntity entity){

		if (entity.getModelId() != null){
			dataModelApplyService.saveDataModelApplyInstance(entity.getModelId(),entity.getModelName(),entity.getId(),entity.getTopicTaskName(),"接入类数据模型");
		}
	}

	public List<JSONObject> jsonStructureData(List<DataModelFieldEntity> dataModelFieldEntities){

		List<JSONObject> jsonObjects = new ArrayList<>();
		for (int i=0;i< dataModelFieldEntities.size();i++){
			JSONObject jsonObject = new JSONObject();
			jsonObject.put("fieldName",dataModelFieldEntities.get(i).getFieldName());
			jsonObject.put("fieldType",dataModelFieldEntities.get(i).getFieldType());
			jsonObject.put("fieldComments",dataModelFieldEntities.get(i).getFieldCnName());
			jsonObjects.add(jsonObject);
		}
		return jsonObjects;
	}

}
